package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
/**
 * @Class:
 * @Author: kongcb
 * @Description: 双随机抽查检查（中数的全量+增量）
 * @Date: 2023/12/5 11:29
 */
public class TRandomInspect {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_random_inspect").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_random_inspect");
        long zs_count = zs_data.count();
        if(zs_count>0){
            zs_data.createOrReplaceTempView("zs_tab");
            //删除数据
            spark.sql("select " +
                    "RECORD_ID,\n" +
                    "DETAIL_ID,\n" +
                    "company_id,\n" +
                    "company_name,\n" +
                    "INSPECT_ORG,\n" +
                    "INSPECT_DATE,\n" +
                    "INSPECT_PLAN_ID,\n" +
                    "INSPECT_PLAN_NAME,\n" +
                    "INSPECT_TASK_ID,\n" +
                    "INSPECT_TASK_NAME,\n" +
                    "INSPECT_TYPE,\n" +
                    "CREDIT_CODE,\n" +
                    "REGNO,\n" +
                    "dates, " +
                    "'-1' as isadd " +
                    "from a_dc_ep_dwi.t_random_inspect  where " +
                    "RECORD_ID in (select a.RECORD_ID from a_dc_ep_ods.zs_t_random_inspect_del as a " +
                    "left join a_dc_ep_ods.zs_t_random_inspect as b on a.RECORD_ID=b.RECORD_ID where b.RECORD_ID is null)")
                    .createOrReplaceTempView("res_delete");
            //优先通过统一社会信用代码作为关联条件来补充company_id
            spark.sql("select a.RECORD_ID,\n" +
                    "a.DETAIL_ID,\n" +
                    "b.company_id,\n" +
                    "a.ENTNAME as company_name,\n" +
                    "a.INSPECT_ORG,\n" +
                    "a.INSPECT_DATE,\n" +
                    "a.INSPECT_PLAN_ID,\n" +
                    "a.INSPECT_PLAN_NAME,\n" +
                    "a.INSPECT_TASK_ID,\n" +
                    "a.INSPECT_TASK_NAME,\n" +
                    "a.INSPECT_TYPE,\n" +
                    "a.CREDIT_CODE,\n" +
                    "a.REGNO,\n" +
                    "a.JOBID as dates " +
                    "from " +
                    "(select * from a_dc_ep_ods.zs_t_random_inspect where length(credit_code)=18) as a " +
                    "left join a_dc_ep_dwi.t_eci_company as b " +
                    "on a.credit_code=b.credit_code")
                    .createOrReplaceTempView("res_tab1");
            //在统一社会信用代码没有补充到id的时候可以在通过注册号来进行补充，但由于注册号会出现很多脏数据，需要对注册号进行筛选。
            spark.sql("select a.RECORD_ID,\n" +
                    "a.DETAIL_ID,\n" +
                    "b.company_id,\n" +
                    "a.ENTNAME as company_name,\n" +
                    "a.INSPECT_ORG,\n" +
                    "a.INSPECT_DATE,\n" +
                    "a.INSPECT_PLAN_ID,\n" +
                    "a.INSPECT_PLAN_NAME,\n" +
                    "a.INSPECT_TASK_ID,\n" +
                    "a.INSPECT_TASK_NAME,\n" +
                    "a.INSPECT_TYPE,\n" +
                    "a.CREDIT_CODE,\n" +
                    "a.REGNO,\n" +
                    "a.JOBID as dates " +
                    "from " +
                    "(select * from a_dc_ep_ods.zs_t_random_inspect where DETAIL_ID not in (select DETAIL_ID from res_tab1) and length(regno)>10) as a " +
                    "left join a_dc_ep_dwi.t_eci_company as b " +
                    "on a.regno=b.no")
                    .createOrReplaceTempView("res_tab2");
            //在统一社会信用代码和注册号都没有关联上的同时，可以通过企业名称进行关联，可能出现错误的情况。（可补充一部分，如果有关联错误的情况，发现了可手动修改）---防止数据缺失。
            spark.sql("select a.RECORD_ID,\n" +
                    "a.DETAIL_ID,\n" +
                    "b.company_id,\n" +
                    "a.ENTNAME as company_name,\n" +
                    "a.INSPECT_ORG,\n" +
                    "a.INSPECT_DATE,\n" +
                    "a.INSPECT_PLAN_ID,\n" +
                    "a.INSPECT_PLAN_NAME,\n" +
                    "a.INSPECT_TASK_ID,\n" +
                    "a.INSPECT_TASK_NAME,\n" +
                    "a.INSPECT_TYPE,\n" +
                    "a.CREDIT_CODE,\n" +
                    "a.REGNO,\n" +
                    "a.JOBID as dates " +
                    "from " +
                    "(select * from a_dc_ep_ods.zs_t_random_inspect where DETAIL_ID not in " +
                    "(select DETAIL_ID from res_tab1 union all select DETAIL_ID from res_tab2) and length(ENTNAME)>5) as a " +
                    "left join a_dc_ep_dwi.t_eci_company as b " +
                    "on regexp_replace(trim(b.company_name),'([(（）)])','') = regexp_replace(trim(a.ENTNAME),'([(（）)])','')")
                    .createOrReplaceTempView("res_tab3");
            //该部分的企业id是没有关联上主表的，可以暂时先置空，后面可以作为补充数据的基础。
            spark.sql("select a.RECORD_ID,\n" +
                    "a.DETAIL_ID,\n" +
                    "'' as company_id,\n" +
                    "a.ENTNAME as company_name,\n" +
                    "a.INSPECT_ORG,\n" +
                    "a.INSPECT_DATE,\n" +
                    "a.INSPECT_PLAN_ID,\n" +
                    "a.INSPECT_PLAN_NAME,\n" +
                    "a.INSPECT_TASK_ID,\n" +
                    "a.INSPECT_TASK_NAME,\n" +
                    "a.INSPECT_TYPE,\n" +
                    "a.CREDIT_CODE,\n" +
                    "a.REGNO,\n" +
                    "a.JOBID as dates " +
                    "from " +
                    "(select * from a_dc_ep_ods.zs_t_random_inspect where DETAIL_ID not in " +
                    "(select DETAIL_ID from res_tab1 union all select DETAIL_ID from res_tab2 union all select DETAIL_ID from res_tab3)) as a ")
                    .createOrReplaceTempView("res_tab4");
            spark.sql("select * from res_tab1 " +
                    "union all " +
                    "select * from res_tab2 " +
                    "union all " +
                    "select * from res_tab3 " +
                    "union all " +
                    "select * from res_tab4")
                    .createOrReplaceTempView("res_tab5");
            //新增数据
            spark.sql("select a.*,'1' as isadd from res_tab5 as a " +
                    "left join a_dc_ep_ods.zs_t_random_inspect_del as b " +
                    "on a.RECORD_ID = b.RECORD_ID where b.RECORD_ID is null").createOrReplaceTempView("res_insert");
            //修改数据
            spark.sql("select a.*,'0' as isadd from res_tab5 as a " +
                    "inner join a_dc_ep_ods.zs_t_random_inspect_del as b " +
                    "on a.RECORD_ID = b.RECORD_ID").createOrReplaceTempView("res_update");

            spark.sql("insert overwrite table a_dc_ep_incr.t_random_inspect (select * from res_insert union all select * from res_update union all select * from res_delete)");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_random_inspect");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
